# -*- coding: utf-8 -*-
# @name             : Excel用例执行
# @author           : OneYoung
# @create           : 2024/07/18
# @version          : v1.0
# @desc             : 执行Excel文件中的测试用例，用例格式需要与模板一致

import json, time, os
from openpyxl.styles import Font
from core.http_request import Request
from core.custom_assert import Assertion
from utils.file_reader import *
from utils.log_record import logger


# 执行Excel文件中的用例
class ExecuteCaseFromFile:
    http = Request()

    def __init__(self):
        self.caseNo = ""
        self.caseName = ""
        self.url = ""
        self.method = ""
        self.header = {}
        self.params = {}
        self.data = {}
        self.json = {}
        self.files = []
        self.assert_info = {}
        self.skip = ""
        self.response = None

    # 从Excel文件中读取测试用例集合
    @staticmethod
    def __read_case_list(excel_name, sheet_name):
        case_list = []
        try:
            # 打开Excel文件
            suffix = excel_name.split('.')[1]  # 截取文件后缀
            if suffix == "xlsx":
                wb = openpyxl.load_workbook(excel_name)
                sheet = wb[sheet_name]
                values = list(sheet.values)
                for value in values:
                    if value[1] != u'case_name':
                        case_list.append(value)
            elif suffix == "xls":
                file = open_workbook(excel_name)
                # 根据名称定位到具体sheet页面
                sheet = file.sheet_by_name(sheet_name)
                # 获取sheet行数据
                rows = sheet.nrows
                for i in range(rows):
                    if sheet.row_values(i)[1] != u'case_name':
                        case_list.append(sheet.row_values(i))
            else:
                raise TypeError("Not invalid Excel file.")
        except FileNotFoundError:
            raise FileNotFoundError(f"The file {excel_name} is not exist.")
        return case_list

    # 读取用例信息
    def __read_case_data(self, case):
        # 用例编号
        self.caseNo = case[0]

        # 用例名称
        self.caseName = case[1]

        # URL 请求地址
        url = case[2]
        if url is None or url == "":
            raise ValueError(f"{self.caseNo}: request url can't be empty.")
        else:
            self.url = url

        # Method 请求方法
        method = case[3]
        if method is None or method == "":
            raise ValueError(f"{self.caseNo}: method can't be empty.")
        else:
            self.method = method

        # Header 请求头
        header = case[4]
        try:
            if header is None or header == "":
                self.header = {}
            else:
                self.header = eval(header.replace('"', "'"))
        except Exception as e:
            self.header = {}
            raise TypeError('The [header] must be a dict.{"key": "value"}', e)

        # Params 参数
        params = case[5]
        try:
            if params is None or params == "":
                self.params = {}
            else:
                self.params = eval(params.replace('"', "'"))
        except Exception as e:
            self.params = {}
            raise TypeError('The [params] must be a dict.{"key": "value"}', e)

        # Data 数据body
        data = case[6]
        try:
            if data is None or data == "":
                self.data = {}
            else:
                self.data = eval(data.replace('"', "'"))
        except Exception as e:
            self.data = {}
            raise TypeError('The [data] must be a dict.{"key": "value"}', e)

        # files 上传文件
        self.filefiles = case[7]

        # Assert_Info 断言信息
        assert_info = case[8]
        try:
            if assert_info is None or assert_info == "":
                self.assert_info = {}
            else:
                self.assert_info = eval(assert_info.replace('"[', "'[").replace(']"', "]'"))
        except Exception as e:
            self.assert_info = {}
            raise TypeError('The [assert_info] must be a dict.{"key": "value"}', e)

        # skip 是否跳过
        self.skip = case[9]

    # 处理常用response属性断言
    def __dispose_response_assert(self, condition):
        if condition == 'status_code':
            assert_value = self.response.status_code
        elif condition == 'content':
            assert_value = self.response.content
        elif condition == 'text':
            assert_value = self.response.text
        elif condition == 'encoding':
            assert_value = self.response.encoding
        elif condition == 'apparent_encoding':
            assert_value = self.response.apparent_encoding
        elif condition == 'request.headers':
            assert_value = self.response.request.headers
        elif condition == 'request.body':
            assert_value = self.response.request.body
        elif condition == 'request.path_url':
            assert_value = self.response.request.path_url
        elif condition == 'request.method':
            assert_value = self.response.request.method
        elif condition == 'url':
            assert_value = self.response.url
        else:
            assert_value = None
        return assert_value

    # 验证断言信息
    def __verify_assert(self):
        if self.assert_info != {}:
            # 获取response响应信息
            try:
                response_json = self.response.json()
            except json.decoder.JSONDecodeError:
                response_json = {}
            # 验证
            for key, value in self.assert_info.items():
                # 处理response常用属性的断言
                response_condition = key.split('.')
                if response_condition[0] == 'response':
                    condition = key.split('response.')[1]
                    res_value = self.__dispose_response_assert(condition)
                else:
                    # 对自定义断言信息做处理，以"->"做分割
                    key_infos = key.split('->')
                    # 对断言信息中的内容做处理，区分读取字符串还是数组
                    assert_key = []
                    for a_key in key_infos:
                        # 如果断言有数组，需要处理，参数值为int
                        if a_key[0] == '[' and a_key[-1] == ']':
                            new_key = a_key[1:-1]
                            try:  # 规避本身包含'[]'符号的参数名不做处理，如果还有其他特殊情况，需要自行规避
                                k = int(new_key)
                            except ValueError:
                                k = a_key
                        else:
                            k = a_key
                        assert_key.append(k)

                    # 验证断言信息
                    res_value = None
                    res_json = response_json  # 因为循环读取会重新赋值，所以执行每个断言前需要重新初始化值
                    for i in range(len(assert_key)):
                        try:
                            res_value = res_json[assert_key[i]]
                            res_json = res_value
                        except KeyError:
                            raise AssertionError(f"AssertionError: '{assert_key[i]}' not found.")
                # 对比结果
                c_assert = Assertion(self.response)
                c_assert.data_equal(res_value, value)

    # 发起请求
    def __call_request(self):
        request_params = dict(
            url=self.url,
            headers=self.header,
            params=self.params,
            data=self.data,
            json=self.json,
            files=self.files
        )
        # 根据method调用请求方法
        if self.method == 'get':
            self.response = self.http.base_request('GET', **request_params)
        elif self.method == 'post':
            self.response = self.http.base_request('POST', **request_params)
        elif self.method == 'put':
            self.response = self.http.base_request('PUT', **request_params)
        elif self.method == 'patch':
            self.response = self.http.base_request('PATCH', **request_params)
        elif self.method == 'delete':
            self.response = self.http.base_request('DELETE', **request_params)

    # 文件检查
    @staticmethod
    def __check_case_file(case_file):
        # 文件名称
        path_split = os.path.split(case_file)
        case_file_path = path_split[0]
        case_file_name = path_split[-1]

        # 判断文件是否存在
        if not os.path.exists(case_file):
            logger.error(f"No such file or directory:{case_file}")
            return True

        # 文件格式必须为.xlsx
        suffix = case_file.split('.')[1]  # 截取文件后缀
        if suffix != "xlsx":
            logger.error(f'{case_file_name} is not support format, please use the .xlsx file format.')
            return True

        # 文件是否打开
        hide_file_name = case_file_path + r"/~$" + case_file_name
        if os.path.exists(hide_file_name):
            logger.error(f"File {case_file_name} is opened, Please closed first.")
            return True

    # 执行用例
    def execute_file_case(self, case_file, sheetName='Sheet1'):
        # 文件检查(存在、xlsx格式、未占用)
        check = self.__check_case_file(case_file)
        if check:
            return

        # 打开用例文件
        wb = openpyxl.load_workbook(case_file)
        sheet = wb[sheetName]

        # 获取用例列表
        cases_list = self.__read_case_list(case_file, sheetName)
        total_num = len(cases_list)  # 用例数量

        # 定义用例执行结果
        pass_num = fail_num = skip_num = 0
        fail_case = []

        # 执行开始时间
        start_time = time.time()
        logger.info("==========START RUN TEST CASE==========")

        # 执行测试
        for i in range(total_num):
            case = cases_list[i]
            # 执行测试前先检查用例参数格式是否有误
            try:
                self.__read_case_data(case)  # 读取用例信息，格式不正确时停止执行
            except Exception as e:
                logger.error(case[0], "Format Error -> ", e)
                return
            # 执行的用例信息
            case_info = str(self.caseNo) + "-" + str(self.caseName)
            # 开始执行
            logger.info(f"Running:{case_info}")

            # 只执行Skip标识不为T(跳过)的用例
            if self.skip not in ['T', 't']:
                try:
                    # 执行用例，发起请求
                    self.__call_request()
                    # 断言
                    self.__verify_assert()

                    # 格式化输出
                    try:
                        res_json = json.dumps(self.response.json(), indent=4, ensure_ascii=False)
                    except json.decoder.JSONDecodeError:
                        res_json = ""
                    pass_num += 1  # 成功数量累加
                    sheet.cell(i + 2, 11).value = 'Pass'  # 回写成功测试结果
                    sheet.cell(i + 2, 11).font = Font(bold=True, color='2C9B37')  # 字体样式
                    sheet.cell(i + 2, 12).value = res_json  # 回写Response信息
                    sheet.cell(i + 2, 13).value = ""  # 清除remark信息

                    msg = f"{case_info}：Test Pass."
                    logger.info(msg)

                except Exception as exception:
                    try:
                        if self.response:
                            res_json = json.dumps(self.response.json(), indent=4, ensure_ascii=False)
                        else:
                            res_json = ""
                    except json.decoder.JSONDecodeError:
                        res_json = self.response.text
                    except Exception as error:
                        res_json = ""
                        exception = error
                    fail_num += 1  # 失败数量累加
                    fail_case.append(self.caseNo)  # 统计失败用例
                    sheet.cell(i + 2, 11).value = 'Fail'  # 回写失败测试结果
                    sheet.cell(i + 2, 11).font = Font(bold=True, color='DD1D2A')  # 字体样式
                    sheet.cell(i + 2, 12).value = res_json  # 回写Response信息
                    sheet.cell(i + 2, 13).value = str(exception)  # 回写失败信息

                    msg = f"{case_info}：Test Fail -> {exception}"
                    logger.error(msg)

            else:
                skip_num += 1  # 跳过执行,数量累加
                msg = f"Skip this case: {case_info}"
                logger.info(msg)

        # 保存测试结果
        wb.save(case_file)

        # 执行结束时间
        end_time = time.time()
        logger.info("==========TEST CASE EXECUTE END==========")
        logger.info(f"Test failed caseNo: {fail_case}\n")

        result = f"\n========== TEST RESULT ==========\n" \
                 f"CASE NUM: {total_num}\n" \
                 f"TOTAL TIME(s): {end_time - start_time}\n" \
                 f"PASS: {pass_num}\n" \
                 f"FAIL: {fail_num}\n" \
                 f"SKIP: {skip_num}\n" \
                 f"FAIL CASE LIST: {fail_case}"
        logger.info(result)


if __name__ == '__main__':
    exec_case = ExecuteCaseFromFile()
    exec_case.execute_file_case("testCase.xlsx")
